package com.cn.goodsadmin.common.observes;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created with IntelliJ IDEA.

 * User: fly
 * Date: 2019-11-15
 * Time: 11:35 下午
 */
@Slf4j
public enum YaKashMessageManager {

    INSTANCE;

    private final ArrayBlockingQueue<Message> messageArrayBlockingQueue;

    private final Map<String, List<IObserver>> obserMap;

    /**
     * 业务的工作线程
     */
    private final ThreadPoolExecutor workThreadPool;

    YaKashMessageManager() {

        messageArrayBlockingQueue = new ArrayBlockingQueue<Message>(10000);

        obserMap = new HashMap<>();
        workThreadPool = new ThreadPoolExecutor
                (10,
                        10,
                        0L,
                        TimeUnit.MILLISECONDS,
                        new ArrayBlockingQueue<Runnable>(200));
        workThreadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    }

    /**
     * 注册观察者
     *
     * @param messageId
     * @param observesr
     */
    public synchronized void registerMessage(String messageId, IObserver observesr) {
        List<IObserver> observesrs = obserMap.get(messageId);
        if (observesrs == null) {
            observesrs = new ArrayList<IObserver>();
            obserMap.put(messageId, observesrs);
        }
        observesrs.add(observesr);
    }

    /**
     * 注销观察者
     *
     * @param messageId
     * @param observesr
     */
    public synchronized void unRegisterMessage(String messageId, IObserver observesr) {
        List<IObserver> observesrs = obserMap.get(messageId);
        if (observesrs != null) {
            observesrs.remove(observesr);
        }
    }

    public void productMessage(Message message) {
        try {
            log.info("productMessage msg");
            boolean flag = messageArrayBlockingQueue.offer(message);
            if (!flag) {
                log.error(" lost message :" + message.messageid);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 消费消息
     */
    public void cosumeMessage() {

        while (true) {
            try {
                final Message message = messageArrayBlockingQueue.take();
                log.info("take come in ==" + message.messageid);
                if (message != null) {
                    String messageid = String.valueOf(message.messageid);
                    List<IObserver> list = obserMap.get(messageid);
                    log.info("IObservers list:" + list);
                    if (list != null && list.size() > 0) {
                        for (final IObserver observesr : list) {
                            Runnable runnable = new Runnable() {
                                @Override
                                public void run() {
                                    observesr.onNoticeMessage(message);
                                }
                            };
                            workThreadPool.execute(runnable);
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 销毁
     */
    public void clearAll() {
        obserMap.clear();
        messageArrayBlockingQueue.clear();
    }
}
